03 蓝图拆分大型Agent项目
上一篇文章最后,我们把所有接口都写在一个app.py里:健康检查、对话、会话管理、错误处理……随着功能增加,这个文件会越来越长,越来越难维护。
你可能会想:把不同功能的路由分到不同文件不就行了?问题是,Flask的@app.route()装饰器需要一个app对象,如果每个文件都import app,就会产生循环引用问题。
Blueprint(蓝图)就是Flask提供的解决方案——先定义路由,后注册到应用。你可以在独立的文件中定义路由,最后在创建应用时把它们组装起来。
打个比方:如果Flask应用是一栋大楼,Blueprint就是每层楼的施工图纸。施工队(路由)在各自的图纸上画好方案,最后交给你统一施工(注册到应用)。
一、创建Blueprint
一个Blueprint就是一个独立的路由模块。创建方式和Flask应用很像:
# routes/chat.py
from flask import Blueprint, request
# 创建Blueprint实例
# 第一个参数是蓝图名称,第二个参数通常是__name__
chat_bp = Blueprint("chat", __name__)
@chat_bp.route("/chat", methods=["POST"])
def chat():
"""对话接口"""
data = request.get_json()
message = data.get("message", "")
return {"reply": f"收到: {message}"}
@chat_bp.route("/chat/models")
def list_models():
"""列出可用模型"""
return {"models": ["deepseek-v4-flash", "deepseek-v3"]}注意看:这里用的是@chat_bp.route()而不是@app.route()。Blueprint只是记录了"我要注册这些路由",但还没真正生效。
1.1 Blueprint构造参数
Blueprint(name, import_name, **options)| 参数 | 说明 | 示例 |
|---|---|---|
name | 蓝图名称,用于端点前缀 | "chat" |
import_name | 通常传__name__,用于定位资源文件 | __name__ |
url_prefix | URL前缀,所有路由都会加上这个前缀 | "/api/v1" |
template_folder | 模板文件夹 | "templates" |
static_folder | 静态文件夹 | "static" |
二、注册Blueprint
定义好Blueprint之后,需要在应用中注册它:
# app.py
from flask import Flask
from routes.chat import chat_bp
app = Flask(__name__)
# 注册蓝图
app.register_blueprint(chat_bp)
if __name__ == "__main__":
app.run(debug=True)注册之后,chat_bp中定义的所有路由才会真正生效。
2.1 URL前缀
注册时可以指定URL前缀,让所有路由都加上统一的路径:
# 所有路由前面自动加上 /api/v1
app.register_blueprint(chat_bp, url_prefix="/api/v1")这样原来定义的/chat就变成了/api/v1/chat,/chat/models变成了/api/v1/chat/models。
前缀也可以在创建Blueprint时就指定:
chat_bp = Blueprint("chat", __name__, url_prefix="/api/v1")两种方式效果一样,选哪种看你的习惯。
三、拆分Agent项目
来看一个实际的Agent项目拆分方案:
my-agent-api/
├── app.py # 应用入口,组装所有蓝图
├── routes/
│ ├── __init__.py
│ ├── chat.py # 对话相关接口
│ ├── session.py # 会话管理接口
│ └── health.py # 健康检查接口
└── errors.py # 错误处理3.1 chat.py — 对话接口
# routes/chat.py
from flask import Blueprint, request, abort
chat_bp = Blueprint("chat", __name__)
@chat_bp.route("/chat", methods=["POST"])
def chat():
"""发送消息,获取Agent回复"""
data = request.get_json()
if not data or "message" not in data:
abort(400, description="缺少message字段")
message = data["message"]
session_id = data.get("session_id", "default")
# 后面会替换成真正的Agent调用
return {
"reply": f"收到: {message}",
"session_id": session_id,
}
@chat_bp.route("/chat/models")
def list_models():
"""列出可用模型"""
return {
"models": [
{"id": "deepseek-v4-flash", "name": "DeepSeek V4 Flash"},
{"id": "deepseek-v3", "name": "DeepSeek V3"},
]
}3.2 session.py — 会话管理
# routes/session.py
from flask import Blueprint, jsonify
session_bp = Blueprint("session", __name__)
@session_bp.route("/session/<session_id>")
def get_session(session_id):
"""获取会话信息"""
return {
"session_id": session_id,
"messages": [],
"created_at": "2026-01-01T00:00:00Z",
}
@session_bp.route("/session/<session_id>", methods=["DELETE"])
def delete_session(session_id):
"""删除会话"""
return {"deleted": session_id}
@session_bp.route("/sessions")
def list_sessions():
"""列出所有会话"""
return {"sessions": []}3.3 health.py — 健康检查
# routes/health.py
from flask import Blueprint
health_bp = Blueprint("health", __name__)
@health_bp.route("/health")
def health():
"""健康检查"""
return {"status": "ok"}
@health_bp.route("/version")
def version():
"""版本信息"""
return {"version": "1.0.0"}3.4 errors.py — 错误处理
# errors.py
from flask import Flask, jsonify
def register_error_handlers(app: Flask):
"""注册全局错误处理器"""
@app.errorhandler(400)
def bad_request(error):
return jsonify({"error": str(error.description)}), 400
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "接口不存在"}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({"error": "服务器内部错误"}), 5003.5 app.py — 组装一切
# app.py
from flask import Flask
from routes.chat import chat_bp
from routes.session import session_bp
from routes.health import health_bp
from errors import register_error_handlers
def create_app():
"""创建Flask应用"""
app = Flask(__name__)
# 注册蓝图
app.register_blueprint(chat_bp, url_prefix="/api")
app.register_blueprint(session_bp, url_prefix="/api")
app.register_blueprint(health_bp)
# 注册错误处理
register_error_handlers(app)
return app
if __name__ == "__main__":
app = create_app()
app.run(debug=True)现在app.py只有十几行,清清楚楚。每个模块各管各的,互不干扰。
最终的API列表:
| 方法 | 路径 | 来源 |
|---|---|---|
| GET | /health | health_bp |
| GET | /version | health_bp |
| POST | /api/chat | chat_bp |
| GET | /api/chat/models | chat_bp |
| GET | /api/session/<id> | session_bp |
| DELETE | /api/session/<id> | session_bp |
| GET | /api/sessions | session_bp |
四、Blueprint的错误处理
Blueprint可以有自己的错误处理器:
chat_bp = Blueprint("chat", __name__)
@chat_bp.errorhandler(429)
def rate_limit_exceeded(error):
"""对话接口的限流错误"""
return {"error": "请求太频繁,请稍后再试"}, 429Blueprint的错误处理器只在该蓝图的视图函数中触发。如果蓝图内部没有匹配的处理器,会向上查找应用级别的处理器。
一个实用的做法:按前缀区分错误格式:
@app.errorhandler(404)
@app.errorhandler(405)
def handle_api_error(ex):
if request.path.startswith("/api/"):
# API接口返回JSON
return jsonify(error=str(ex)), ex.code
else:
# 其他页面返回默认HTML
return ex五、Blueprint的URL生成
用url_for生成Blueprint中的URL时,需要加上蓝图名前缀:
from flask import url_for
# 格式:蓝图名.函数名
url_for("chat.chat") # → /api/chat
url_for("session.get_session", session_id="abc") # → /api/session/abc
url_for("health.health") # → /health在Blueprint内部可以省略蓝图名,用.函数名:
@chat_bp.route("/chat")
def chat():
# 在chat_bp内部跳转到其他端点
models_url = url_for(".list_models") # → /api/chat/models
return {"models_url": models_url}六、嵌套Blueprint
Blueprint可以注册到另一个Blueprint上,实现更细粒度的模块划分:
# 父蓝图
api_bp = Blueprint("api", __name__, url_prefix="/api")
# 子蓝图
chat_bp = Blueprint("chat", __name__)
session_bp = Blueprint("session", __name__)
# 子蓝图注册到父蓝图
api_bp.register_blueprint(chat_bp, url_prefix="/chat")
api_bp.register_blueprint(session_bp, url_prefix="/session")
# 父蓝图注册到应用
app.register_blueprint(api_bp)最终的URL:
| 路径 | 说明 |
|---|---|
/api/chat/ | chat_bp的/路由 |
/api/chat/models | chat_bp的/models路由 |
/api/session/<id> | session_bp的/<id>路由 |
用url_for时需要完整的嵌套路径:
url_for("api.chat.chat") # → /api/chat/
url_for("api.session.get_session", session_id="abc") # → /api/session/abc这种方式适合API版本管理——v1和v2各自是一个父蓝图,下面挂不同的子蓝图。
七、总结
Blueprint是Flask组织大型项目的核心机制:
- 先定义后注册:在独立文件中定义路由,最后组装到应用
- URL前缀:统一管理API路径,如
/api/v1 - 模块化:每个Blueprint负责一个功能领域
- 错误处理:Blueprint可以有自己的错误处理器
- 嵌套:Blueprint可以嵌套,适合API版本管理
有了Blueprint,你的Agent项目就可以从一个app.py演进成一个结构清晰的多模块项目。
在下一篇文章中,我们将学习流式响应——这是Agent API最重要的特性之一,让用户能看到Agent逐字输出的效果。